﻿using NetMQ;
using NetMQ.Sockets;
using Newtonsoft.Json;
using System;
using System.Text;

namespace ParallelLib.Micro
{
    internal class NetMQClient : IDisposable
    {
        private NetMQPoller _poller;
        private SubscriberSocket _subscriber;

        public event Action<NetMQClient, string, string> MessageReceived;

        public bool IsRunning
        {
            get { return _poller != null && _poller.IsRunning; }
        }

        public BeaconNode Node { get; private set; }

        public string ServiceAddress { get; private set; }

        public MQConfig ServiceConfig { get; private set; }

        public void Dispose()
        {
            throw new NotImplementedException();
        }

        public string GetWebServiceUri()
        {
            if (this.ServiceConfig == null)
                return string.Empty;
            return string.Format("http://{0}:{1}", this.ServiceAddress, this.ServiceConfig.WebPort);
        }

        public bool Start(BeaconNode node)
        {
            if (node == null || string.IsNullOrEmpty(node.Arguments)) return false;

            this.Node = node;
            try
            {
                ServiceConfig = JsonConvert.DeserializeObject<MQConfig>(node.Arguments);
                if (ServiceConfig == null || ServiceConfig.PubPort == 0)
                    return false;
                return this.Start(node.GetHostAddress(), ServiceConfig.PubPort);
            }
            catch (Exception)
            {
                return false;
            }
        }

        public void Stop()
        {
            if (_subscriber != null)
            {
                _subscriber.ReceiveReady -= _subscriber_ReceiveReady;
            }
            if (_poller != null)
            {
                _poller.Stop();
                _poller.Remove(_subscriber);
            }
            this.ServiceAddress = string.Empty;
            this.ServiceConfig = null;
        }

        private void _subscriber_ReceiveReady(object sender, NetMQSocketEventArgs e)
        {
            var msg = _subscriber.ReceiveMultipartMessage();
            if (msg == null || msg.FrameCount != 2)
                return;
            string msgKey = msg[0].ConvertToString();
            string value = msg[1].ConvertToString(Encoding.UTF8);
            MessageReceived?.Invoke(this, msgKey, value);
        }

        private string GetZMQPubUri(string serviceAddress = null, int pubPort = 0)
        {
            return string.Format("tcp://{0}:{1}",
                string.IsNullOrEmpty(serviceAddress) ? this.ServiceAddress : serviceAddress,
                pubPort != 0 ? pubPort : (this.ServiceConfig == null ? 0 : this.ServiceConfig.PubPort));
        }

        private void OnMessageReceived(string key, string message)
        {
            if (this.MessageReceived != null)
                MessageReceived(this, key, message);
        }

        private bool Start(string address, int port)
        {
            ServiceAddress = address;

            _subscriber = new SubscriberSocket();
            _subscriber.Connect(GetZMQPubUri(address, port));
            _subscriber.ReceiveReady += _subscriber_ReceiveReady;
            _subscriber.SubscribeToAnyTopic();

            _poller = new NetMQPoller { _subscriber };

            _poller.RunAsync();

            return _poller.IsRunning;
        }
    }

    internal class NetMQServer : IDisposable
    {
        private NetMQPoller _poller;
        private PublisherSocket _publisher;

        public event Action<NetMQServer, string, string> MessageReceived;

        public bool IsRunning
        {
            get { return _poller != null && _poller.IsRunning; }
        }

        public BeaconNode Node { get; private set; }

        public string ServiceAddress { get; private set; }

        public MQConfig ServiceConfig { get; private set; }

        public void Dispose()
        {
            throw new NotImplementedException();
        }

        public string GetWebServiceUri()
        {
            if (this.ServiceConfig == null) return string.Empty;

            return string.Format("http://{0}:{1}", this.ServiceAddress, this.ServiceConfig.WebPort);
        }

        public bool Start(BeaconNode node)
        {
            if (node == null || string.IsNullOrEmpty(node.Arguments))
                return false;
            this.Node = node;
            try
            {
                ServiceConfig = JsonConvert.DeserializeObject<MQConfig>(node.Arguments);
                if (ServiceConfig == null || ServiceConfig.PubPort == 0)
                    return false;
                return this.Start(node.GetHostAddress(), ServiceConfig.PubPort);
            }
            catch (Exception)
            {
                return false;
            }
        }

        public void Stop()
        {
            if (_publisher != null)
            {
                _publisher.ReceiveReady -= _publisher_ReceiveReady;
                _publisher.SendReady -= _publisher_SendReady;
            }
            if (_poller != null)
            {
                _poller.Stop();
                _poller.Remove(_publisher);
            }
            this.ServiceAddress = string.Empty;
            this.ServiceConfig = null;
        }

        private void _publisher_ReceiveReady(object sender, NetMQSocketEventArgs e)
        {
            var msg = _publisher.ReceiveMultipartMessage();
            if (msg == null || msg.FrameCount != 2)
                return;
            string msgKey = msg[0].ConvertToString();
            string value = msg[1].ConvertToString(Encoding.UTF8);
            MessageReceived?.Invoke(this, msgKey, value);
        }

        private string GetZMQPubUri(string serviceAddress = null, int pubPort = 0)
        {
            return string.Format("tcp://{0}:{1}",
                string.IsNullOrEmpty(serviceAddress) ? this.ServiceAddress : serviceAddress,
                pubPort != 0 ? pubPort : (this.ServiceConfig == null ? 0 : this.ServiceConfig.PubPort));
        }

        private bool Start(string address, int port)
        {
            ServiceAddress = address;
            _publisher = new PublisherSocket();
            _publisher.Connect(GetZMQPubUri(address, port));
            _publisher.ReceiveReady += _publisher_ReceiveReady;
            _publisher.SendReady += _publisher_SendReady;
            _poller = new NetMQPoller { _publisher };
            _poller.RunAsync();
            return _poller.IsRunning;
        }

        private void _publisher_SendReady(object sender, NetMQSocketEventArgs e)
        {
            //throw new NotImplementedException();
        }
    }

    internal class NetMQXClient : IDisposable
    {
        private NetMQPoller _poller;
        private XSubscriberSocket _subscriber;

        public event Action<NetMQXClient, string, string> MessageReceived;

        public bool IsRunning
        {
            get { return _poller != null && _poller.IsRunning; }
        }

        public BeaconNode Node { get; private set; }

        public string ServiceAddress { get; private set; }

        public MQConfig ServiceConfig { get; private set; }

        public void Dispose()
        {
            throw new NotImplementedException();
        }

        public string GetWebServiceUri()
        {
            if (this.ServiceConfig == null) return string.Empty;

            return string.Format("http://{0}:{1}", this.ServiceAddress, this.ServiceConfig.WebPort);
        }

        public bool Start(BeaconNode node)
        {
            if (node == null || string.IsNullOrEmpty(node.Arguments)) 
                return false;
            this.Node = node;
            try
            {
                ServiceConfig = JsonConvert.DeserializeObject<MQConfig>(node.Arguments);
                if (ServiceConfig == null || ServiceConfig.PubPort == 0)
                    return false;
                return this.Start(node.GetHostAddress(), ServiceConfig.PubPort);
            }
            catch (Exception)
            {
                return false;
            }
        }

        public void Stop()
        {
            if (_subscriber != null)
            {
                _subscriber.ReceiveReady -= _subscriber_ReceiveReady;
            }
            if (_poller != null)
            {
                _poller.Stop();
                _poller.Remove(_subscriber);
            }
            this.ServiceAddress = string.Empty;
            this.ServiceConfig = null;
        }

        private void _subscriber_ReceiveReady(object sender, NetMQSocketEventArgs e)
        {
            var msg = _subscriber.ReceiveMultipartMessage();
            if (msg == null || msg.FrameCount != 2)
            {
                return;
            }
            string msgKey = msg[0].ConvertToString();
            string value = msg[1].ConvertToString(Encoding.UTF8);
            MessageReceived?.Invoke(this, msgKey, value);
        }

        private string GetZMQPubUri(string serviceAddress = null, int pubPort = 0)
        {
            return string.Format("tcp://{0}:{1}",
                string.IsNullOrEmpty(serviceAddress) ? this.ServiceAddress : serviceAddress,
                pubPort != 0 ? pubPort : (this.ServiceConfig == null ? 0 : this.ServiceConfig.PubPort));
        }

        private bool Start(string address, int port)
        {
            ServiceAddress = address;
            _subscriber = new XSubscriberSocket();
            _subscriber.Connect(GetZMQPubUri(address, port));
            _subscriber.ReceiveReady += _subscriber_ReceiveReady;
            _subscriber.SubscribeToAnyTopic();
            _poller = new NetMQPoller { _subscriber };
            _poller.RunAsync();
            return _poller.IsRunning;
        }
    }

    internal class NetMQXServer : IDisposable
    {
        private NetMQPoller _poller;
        private XPublisherSocket _publisher;

        public event Action<NetMQXServer, string, string> MessageReceived;

        public bool IsRunning
        {
            get { return _poller != null && _poller.IsRunning; }
        }

        public BeaconNode Node { get; private set; }

        public string ServiceAddress { get; private set; }

        public MQConfig ServiceConfig { get; private set; }

        public void Dispose()
        {
            throw new NotImplementedException();
        }

        public string GetWebServiceUri()
        {
            if (this.ServiceConfig == null) 
                return string.Empty;
            return string.Format("http://{0}:{1}", this.ServiceAddress, this.ServiceConfig.WebPort);
        }

        public bool Start(BeaconNode node)
        {
            if (node == null || string.IsNullOrEmpty(node.Arguments)) 
                return false;
            this.Node = node;
            try
            {
                ServiceConfig = JsonConvert.DeserializeObject<MQConfig>(node.Arguments);
                if (ServiceConfig == null || ServiceConfig.PubPort == 0) return false;

                return this.Start(node.GetHostAddress(), ServiceConfig.PubPort);
            }
            catch (Exception ex)
            {
                return false;
            }
        }

        public void Stop()
        {
            if (_publisher != null)
            {
                _publisher.ReceiveReady -= _publisher_ReceiveReady;
            }

            if (_poller != null)
            {
                _poller.Stop();
                _poller.Remove(_publisher);
            }

            this.ServiceAddress = string.Empty;
            this.ServiceConfig = null;
        }

        private void _publisher_ReceiveReady(object sender, NetMQSocketEventArgs e)
        {
            var msg = _publisher.ReceiveMultipartMessage();
            if (msg == null || msg.FrameCount != 2)
            {
                return;
            }
            string msgKey = msg[0].ConvertToString();
            string value = msg[1].ConvertToString(Encoding.UTF8);
            MessageReceived?.Invoke(this, msgKey, value);
        }

        private void _publisher_SendReady(object sender, NetMQSocketEventArgs e)
        {
            //throw new NotImplementedException();
        }

        private string GetZMQPubUri(string serviceAddress = null, int pubPort = 0)
        {
            return string.Format("tcp://{0}:{1}",
                string.IsNullOrEmpty(serviceAddress) ? this.ServiceAddress : serviceAddress,
                pubPort != 0 ? pubPort : (this.ServiceConfig == null ? 0 : this.ServiceConfig.PubPort));
        }

        private bool Start(string address, int port)
        {
            ServiceAddress = address;
            _publisher = new XPublisherSocket();
            _publisher.Connect(GetZMQPubUri(address, port));
            _publisher.ReceiveReady += _publisher_ReceiveReady;
            _publisher.SendReady += _publisher_SendReady;
            _poller = new NetMQPoller { _publisher };
            _poller.RunAsync();
            return _poller.IsRunning;
        }
    }
}